package cn.ivicar.cloud.engine;

import cn.ivicar.cloud.cache.*;
import cn.ivicar.cloud.gps.GISUtil3;
import cn.ivicar.cloud.gps.MessageBean;
import cn.ivicar.cloud.guice.IvicarGuice;
import cn.ivicar.cloud.influxdb.InfluxdbImpl;
import cn.ivicar.cloud.influxdb.InfluxdbIntf;
import cn.ivicar.cloud.utils.DataType;
import cn.ivicar.cloud.utils.EventType;
import cn.ivicar.cloud.utils.IovProperties;
import cn.ivicar.redis.JedisClusterFactory;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import redis.clients.jedis.JedisCluster;

import java.math.BigDecimal;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

public class RuleEngine {
    private static Logger LOGGER = Logger.getLogger(RuleEngine.class);

    public static String INFLUXDB_NAME = "ivicar_iov";
    public static String IVICAR_EVENT_CHANNEL = "ivicar_event";
    public static InfluxdbIntf influxdb;

    private static ExecutorService pool;
    private static int WORKER_NUMBER = 100;
    private static Semaphore semaphore = new Semaphore(WORKER_NUMBER);

    private static Activity activity = IvicarGuice.injector.getInstance(Activity.class);
    private static Interaction interaction = IvicarGuice.injector.getInstance(Interaction.class);
    private static Shadow shadow = IvicarGuice.injector.getInstance(Shadow.class);

    static {
        try {
            String dbUrl = IovProperties.getProperties().getProperty("influxdb.url", "http://127.0.0.1:8086");
            String username = IovProperties.getProperties().getProperty("influxdb.username", "ivicar");
            String password = IovProperties.getProperties().getProperty("influxdb.password", "ivicar");

            influxdb = new InfluxdbImpl(dbUrl, username, password);
        } catch (Exception e) {
            e.printStackTrace();
            LOGGER.error(String.format("时序数据库初始化失败。 ErrorMsg:[%s]", e.getMessage()));
        }
    }

    public static void process(String topic, String message) {
        try {
            semaphore.acquire();
            ApplyRuleThread thread = new ApplyRuleThread(message);

            if (pool == null) {
                pool = Executors.newFixedThreadPool(WORKER_NUMBER);
            }

            pool.execute(thread);
        } catch (InterruptedException e) {
            LOGGER.error(String.format("消息处理出现Exception，ErrorMsg:[%s], Message:[%s], Topic:[%s]", e.getMessage(), message, topic));
        } catch (Exception e) {
            LOGGER.error(String.format("消息处理出现Exception，ErrorMsg:[%s], Message:[%s], Topic:[%s]", e.getMessage(), message, topic));
        } finally {
            semaphore.release();
        }
    }

    protected static class ApplyRuleThread extends Thread {
        private String message;

        public ApplyRuleThread(String message) {
            this.message = message;
        }

        @Override
        public void run() {
            try {
                JSONObject json = JSONObject.fromObject(message);
                String clientId = json.getString("id");
                DataType type = DataType.valueOf(json.getString("type").toUpperCase());
                String sn = json.getString("sn");

                if (StringUtils.equals(interaction.getSn(clientId), sn)) {
                    LOGGER.warn(String.format("忽略重复上报的数据：%s", message));
                } else {
                    activity.updateActivity(clientId, type);
                    interaction.updateInteraction(clientId, type);
                    interaction.setInteraction(clientId, new Date().getTime());
                    interaction.setSn(clientId, sn);
                    if (type == DataType.INI) {
                        JSONObject ini = json.getJSONObject("data");
                        shadow.setShadow(clientId, ini, type);
                        interaction.traceDirty(clientId, new Date(ini.getLong("time")));
                        influxdb.write(INFLUXDB_NAME, clientId, type, ini);
                    } else if (type == DataType.GPS) {
                        JSONArray dataArray = json.getJSONArray("data");

                        // 高德地图坐标偏转
                        GISUtil3 gis = new GISUtil3();
                        for (int i = 0; i < dataArray.size(); i++) {
                            MessageBean params = new MessageBean();
                            double lat = Double.valueOf(dataArray.getJSONObject(i).getString("lat"));
                            double lng = Double.valueOf(dataArray.getJSONObject(i).getString("lng"));
                            params.setLatitude(lat);
                            params.setLongitude(lng);
                            gis.getDeflexedCoordinate(params);
                            BigDecimal amapLat = new BigDecimal(params.getAmapLat()).setScale(6, BigDecimal.ROUND_HALF_UP);
                            BigDecimal amapLng = new BigDecimal(params.getAmapLng()).setScale(6, BigDecimal.ROUND_HALF_UP);
                            dataArray.getJSONObject(i).put("amapLat", amapLat.toPlainString());
                            dataArray.getJSONObject(i).put("amapLng", amapLng.toPlainString());
                        }

                        JSONObject lastElement = dataArray.getJSONObject(dataArray.size() - 1);
                        shadow.setShadow(clientId, lastElement, type);
                        interaction.traceDirty(clientId, new Date(lastElement.getLong("time")));
                        influxdb.write(INFLUXDB_NAME, clientId, type, dataArray);
                    } else if (type == DataType.OBD) {
                        JSONArray dataArray = json.getJSONArray("data");
                        JSONObject snapshot = shadow.getShadow(clientId, type);
                        int prevAcc = 0;
                        if (snapshot != null && snapshot.containsKey("acc")) {
                            prevAcc = snapshot.getInt("acc");
                        }

                        // 报警处理
                        JSONObject element;
                        for (int i = 0; i < dataArray.size(); i++) {
                            element = dataArray.getJSONObject(i);
                            int currAcc = 0;
                            if (element.containsKey("acc")) {
                                currAcc = element.getInt("acc");
                            }

                            if (prevAcc != currAcc) {
                                // 点火、熄火处理
                                JedisCluster cluster = JedisClusterFactory.getDefaultJedisCluster();
                                JSONObject accEvent = new JSONObject();
                                accEvent.put("time", element.get("time"));

                                JSONObject gps = shadow.getShadow(clientId, DataType.GPS);
                                accEvent.put("gps", gps);

                                EventType eventType = EventType.NONE;
                                if (prevAcc == 0 && currAcc == 1) {
                                    eventType = EventType.ACC_ON;
                                }

                                if (prevAcc == 1 && currAcc == 0) {
                                    eventType = EventType.ACC_OFF;
                                }

                                accEvent.put("event", eventType.getValue());

                                influxdb.write(INFLUXDB_NAME, clientId, DataType.EVENT, accEvent);

                                accEvent.put("clientId", clientId);
                                accEvent.put("info", "");
                                cluster.lpush(IVICAR_EVENT_CHANNEL, accEvent.toString());
                            }

                            snapshot = element;
                            prevAcc = currAcc;
                        }

                        shadow.setShadow(clientId, snapshot, type);
                        interaction.traceDirty(clientId, new Date(snapshot.getLong("time")));
                        influxdb.write(INFLUXDB_NAME, clientId, type, dataArray);
                    } else if (type == DataType.EVENT) {
                        JedisCluster cluster = JedisClusterFactory.getDefaultJedisCluster();
                        JSONArray dataArray = json.getJSONArray("data");
                        for (int i = 0; i < dataArray.size(); i++) {
                            JSONObject element = dataArray.getJSONObject(i);
                            JSONObject event = new JSONObject();
                            event.put("event", element.get("category"));
                            event.put("clientId", clientId);
                            event.put("info", element.get("info"));
                            event.put("time", element.get("time"));

                            JSONObject gps = shadow.getShadow(clientId, DataType.GPS);
                            event.put("gps", gps);

                            cluster.lpush(IVICAR_EVENT_CHANNEL, event.toString());
                        }

                        influxdb.write(INFLUXDB_NAME, clientId, type, dataArray);
                    } else {
                        LOGGER.error("未知的数据类型：" + type);
                    }
                }
            } catch (Exception e) {
                LOGGER.error(String.format("数据解析处理线程出现Exception。ErrorMsg:[%s], Message:[%s]", e.getMessage(), message));
            }
        }
    }
}
